package rhc.discribute.node.connecter.impl;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;

import org.apache.zookeeper.CreateMode;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;

import rhc.discribute.common.ConfigConstants;
import rhc.discribute.node.commander.Commander;
import rhc.discribute.node.connecter.BasicConnecter;
import rhc.discribute.node.connecter.impl.zookper.ZKExecute;
import rhc.discribute.node.connecter.impl.zookper.ZookperLockImpl;
import rhc.discribute.node.connecter.impl.zookper.ZookperOperate;
import rhc.discribute.node.host.Host;
import rhc.discribute.node.nodeManage.InnerNodeManage;
import rhc.discribute.node.sendMessage.SendMessageCallback;
import rhc.discribute.node.setting.Setting;
import rhc.discribute.signal.Signal;
import rhc.discribute.signal.signalInterface.ReceiveMsgSignal;
import rhc.discribute.util.SystemCloseHookUtil;

/**
 * 通过zookper连接
 * 
 * @author rhc
 *
 */
public class ZookperConnecter extends BasicConnecter {

	/**
	 * 当某一个节点创建了该节点，表示这个节点成为主机
	 */
	public static final String MASTER_HOST_EPHEMERAL_PATH = "/master_host_ephemeral";

	/**
	 * 在线的节点会在该节点下创建临时节点，当临时节点消失表示这个节点离开
	 */
	public static final String HOST_PERSISTENT_PATH = "/host_persistent";

	/**
	 * 消息永久节点，给某一个节点发消息都会在该节点下创建临时节点
	 */
	public static final String MESSAGE_PERSISTENT_PATH = "/message_persistent";

	private ZookperOperate zkOperate;
	
	private ZookperExecute zkExecute;
	
	private ZookperLock zkLock;

	public ZookperConnecter(Commander commander, Setting setting, InnerNodeManage nodeManage) throws Exception {
		super(commander, setting, nodeManage);
	}

	@Override
	public void connect() throws Exception {
		connectZK(setting);
		zkExecute=new ZKExecute(this,zkOperate,nodeManage);
		
		zkLock=new ZookperLockImpl(this,zkOperate);
		
		addMasterHostWatch(setting);
		createMessagePersistentPath();
		zkExecute.addMessageWatch();
		
		addHostNodeJoinWatch(setting);
		zkExecute.hostChange();
	}

	/**
	 * 连接zookeeper
	 * 
	 * @param setting
	 * @throws IOException
	 * @throws InterruptedException
	 */
	private void connectZK(Setting setting) throws IOException, InterruptedException {
		String host = (String) setting.get(ConfigConstants.ZK_HOST);
		zkOperate = new ZookperOperate();
		zkOperate.connectZookeeper(host);
		SystemCloseHookUtil.addSystemCloseHook(new Runnable() {

			@Override
			public void run() {
				try {
					zkOperate.closeConnect();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}

		});
	}

	/**
	 * @param setting
	 * @throws Exception
	 */
	private void addHostNodeJoinWatch(Setting setting) throws Exception {
		createHostPersistentPath();
		createHostMasterEphemeralPath(setting);
		zkExecute.addHostPersistentChildWatch();
	}

	/**
	 * 创建master永久节点
	 * 
	 * @throws Exception
	 * @throws UnsupportedEncodingException
	 */
	private void createHostPersistentPath() throws UnsupportedEncodingException, Exception {
		zkOperate.createPath(HOST_PERSISTENT_PATH, HOST_PERSISTENT_PATH.getBytes("UTF-8"), true, CreateMode.PERSISTENT);
	}

	/**
	 * 创建当前节点master临时节点
	 * 
	 * @throws Exception
	 */
	private void createHostMasterEphemeralPath(Setting setting) throws Exception {
		String localLabel = setting.getProperty(ConfigConstants.LOCAL_LABEL);
		String hostChildPath = getHostEphemeralPath(nodeManage.getCurrentHost());
		zkOperate.createPath(hostChildPath, localLabel.getBytes("UTF-8"), false, CreateMode.EPHEMERAL);
	}

	/**
	 * 获取临时节点路径
	 * 
	 * @param host
	 * @return
	 * @throws Exception
	 */
	public String getHostEphemeralPath(Host host) throws Exception {
		return HOST_PERSISTENT_PATH + "/" + host;
	}

	/**
	 * 创建消息永久父节点
	 * 
	 * @throws UnsupportedEncodingException
	 * @throws Exception
	 */
	private void createMessagePersistentPath() throws UnsupportedEncodingException, Exception {
		zkOperate.createPath(MESSAGE_PERSISTENT_PATH, MESSAGE_PERSISTENT_PATH.getBytes("UTF-8"), true,
				CreateMode.PERSISTENT);
	}

	/**
	 * 获取消息临时节点路径，其它机器发消息时会尝试创建该路径，不成功时会等待路径小时后再创建
	 * 
	 * @param host
	 * @return
	 */
	public String getMessageEphemeralPath(Host host) {
		return MESSAGE_PERSISTENT_PATH + "/" + host;
	}

	/**
	 * 从生成节点路径得到Host对象
	 * 
	 * @param path
	 * @return
	 */
	public Host fromHostEphemeralPathGetHost(String path) {
		path = path.replaceFirst(".*/", "");
		return Host.fromPathGetHost(path);
	}

	/**
	 * 设置当前节点数据
	 * 
	 * @param message
	 * @throws Exception
	 */
	public void setCurrentHostData(String message) throws Exception {
		String path = this.getHostEphemeralPath(nodeManage.getCurrentHost());
		byte[] data = message.getBytes("UTF-8");
		zkOperate.setPathData(path, data);
	}

	

	/**
	 * 增加主节点数据监控，当数据更新表示主节点更新
	 * 
	 * @param setting
	 * @throws Exception
	 */
	public void addMasterHostWatch(Setting setting) throws Exception {
		zkExecute.addMainNodeWatch();
		zkExecute.mainNodeDataChange();
	}

	@Override
	public void sendMessageToHost(Host host, String message,SendMessageCallback callback,boolean wait) throws Exception {
		if (!nodeManage.hostExists(host)) {
			throw new IllegalArgumentException(host + " 不存在");
		}
		message = createSendMessage(message);

		zkExecute.sentMessageToHost(host, message, 1,callback,wait);
	}

	/**
	 * 将字符串转为byte
	 * 
	 * @param str
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	public byte[] getByteData(String str) throws UnsupportedEncodingException {
		if (null == str || str.trim().length() == 0) {
			throw new NullPointerException();
		}

		return str.getBytes("UTF-8");
	}

	/**
	 * 将byte转为string
	 * 
	 * @param data
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	public String getStrData(byte[] data) throws UnsupportedEncodingException {
		if (null == data || data.length == 0) {
			throw new NullPointerException();
		}

		return new String(data, "UTF-8");
	}

	/**
	 * 触发接收消息事件
	 * 
	 * @param message
	 */
	public void touchRecevieMsg(String message) {
		JSONObject json = JSONObject.parseObject(message);
		String hostStr = json.getString("host");
		String msg = json.getString("message");
		Host host = Host.fromPathGetHost(hostStr);

		Map<String, Object> signalObject = new HashMap<String, Object>(2);
		signalObject.put(ReceiveMsgSignal.HOST, host);
		signalObject.put(ReceiveMsgSignal.MESSAGE, msg);
		commander.getSignalManage().touchSignal(Signal.RECEIVE_MSG, signalObject);
	}

	/**
	 * 触发消息发送事件
	 * 
	 * @param message
	 */
	public void touchSendMsg(String message, Host toHost, SendMessageCallback callback) {
		JSONObject json = JSONObject.parseObject(message);
		String hostStr = json.getString("host");
		String msg = json.getString("message");
		Host host = Host.fromPathGetHost(hostStr);
		logger.debug(host+" 发送到 "+toHost+" 的消息 "+msg+" 已经发送");
		if(null != callback){
			callback.send(host, toHost, msg);
		}
	}

	/**
	 * 触发消息成功事件
	 * 
	 * @param message
	 */
	public void touchSendSuccessMsg(String message, Host toHost, SendMessageCallback callback) {
		JSONObject json = JSONObject.parseObject(message);
		String hostStr = json.getString("host");
		String msg = json.getString("message");
		Host host = Host.fromPathGetHost(hostStr);
		logger.debug(host+" 发送到 "+toHost+" 的消息 "+msg+" 发送成功");
		if(null != callback){
			callback.receiveSuccess(host, toHost, msg);
		}
	}
	
	/**触发消息发送失败事件
	 * @param message
	 * @param toHost
	 */
	public void touchSendFailMsg(String message, Host toHost, SendMessageCallback callback) {
		JSONObject json = JSONObject.parseObject(message);
		String hostStr = json.getString("host");
		String msg = json.getString("message");
		Host host = Host.fromPathGetHost(hostStr);
		logger.error(host+" 发送到 "+toHost+" 的消息 "+msg+" 发送失败");
		if(null != callback){
			callback.receiveFail(host, toHost, msg);
		}
	}

	/**
	 * 将本机节点写入消息，让接收消息的节点知道发送消息的是哪个节点
	 * 
	 * @param message
	 * @return
	 * @throws JSONException
	 * @throws Exception
	 */
	public String createSendMessage(String message) throws JSONException, Exception {
		JSONObject json = new JSONObject();
		json.put("host", nodeManage.getCurrentHost().toString());
		json.put("message", message);

		return json.toString();
	}

	@Override
	public ZKPathLockData tryAcquire(String lockPath) throws Exception {
		return zkLock.tryAcquire(lockPath);
	}

	@Override
	public ZKPathLockData tryAcquire(String lockPath, long waitSeconds) throws Exception {
		return zkLock.tryAcquire(lockPath, waitSeconds);
	}

	@Override
	public ZKPathLockData acquire(String lockPath) throws Exception {
		return zkLock.acquire(lockPath);
	}

	@Override
	public void release(ZKPathLockData lockData) throws Exception {
		zkLock.release(lockData);
	}

}
